package com.example.testpulsar.mq.factory;

import com.example.testpulsar.mq.configforpulsar.PulsarConfiguration;
import com.example.testpulsar.mq.configforpulsar.PulsarProperties;
import com.example.testpulsar.mq.constant.MqConstant;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class ConsumerFactory {

    @Resource
    PulsarClient client;

    @Resource
    PulsarProperties pulsarProperties;

    public void createWXGZHHistoryListAcqConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.ACQUISITION_HISTORY_LIST_TOPIC,
                MqConstant.WXGZH_PREFIX + MqConstant.ACQUISITION_HISTORY_LIST_SUBSCRIPTION, SubscriptionType.Exclusive);
    }

    public void createWXGZHHistoryListDealConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.DATA_DEAL_HISTORY_LIST_TOPIC,
                MqConstant.WXGZH_PREFIX + MqConstant.DATA_DEAL_HISTORY_LIST_SUBSCRIPTION, SubscriptionType.Exclusive);
    }

    public void createWXGZHContentAcqConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.ACQUISITION_CONTENT_TOPIC,
                MqConstant.WXGZH_PREFIX + MqConstant.ACQUISITION_CONTENT_SUBSCRIPTION, SubscriptionType.Exclusive);
    }

    public void createWXGZHContentDealConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WXGZH, MqConstant.DATA_DEAL_CONTENT_TOPIC,
                MqConstant.WXGZH_PREFIX + MqConstant.DATA_DEAL_CONTENT_SUBSCRIPTION, SubscriptionType.Exclusive);
    }


    public void createWBHistoryListAcqConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.ACQUISITION_HISTORY_LIST_TOPIC,
                MqConstant.WB_PREFIX + MqConstant.ACQUISITION_HISTORY_LIST_SUBSCRIPTION, SubscriptionType.Exclusive);
    }

    public void createWBHistoryListDealConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.DATA_DEAL_HISTORY_LIST_TOPIC,
                MqConstant.WB_PREFIX + MqConstant.DATA_DEAL_HISTORY_LIST_SUBSCRIPTION, SubscriptionType.Exclusive);
    }

    public void createWBContentAcqConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.ACQUISITION_CONTENT_TOPIC,
                MqConstant.WB_PREFIX + MqConstant.ACQUISITION_CONTENT_SUBSCRIPTION, SubscriptionType.Exclusive);
    }

    public void createWBContentDealConsumer(MessageListener<String> messageListener) throws PulsarClientException {
        PulsarConfiguration.getConsumer(client, pulsarProperties, messageListener, true,
                "public", MqConstant.NAMESPACE_WB, MqConstant.DATA_DEAL_CONTENT_TOPIC,
                MqConstant.WB_PREFIX + MqConstant.DATA_DEAL_CONTENT_SUBSCRIPTION, SubscriptionType.Exclusive);
    }

}
